package ex.tableapi;

import ex.util.Top2;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/tableapi/#groupby-window-aggregation
 * <p>
 * 更换jar ：  https://mvnrepository.com/artifact/org.apache.flink/flink-connector-jdbc/3.1.1-1.17
 */
public class ExFunction extends ApiFrame {

    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExFunction demo = new ExFunction();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");
        //3.调用UDF进行查询转换
        String windowAggQuery =
                "SELECT a,COUNT(b) AS cnt " +
                "FROM orders a group by a";

        Table aggTable = demo.tEnv.sqlQuery(windowAggQuery);

        aggTable.printSchema();

        Table result = demo.exFlatAggregate(aggTable);
        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }


    /**
     * see https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/functions/udfs/
     * @param orders
     * @return
     */
    private Table exInnerJoinWithTableFunction( Table orders) {
//       等同call(SplitFunction.class 等同 tableEnv.registerFunction("split", split);
        Table result = orders.joinLateral(call(SplitFunction.class, $("f")))
                .select($("f"), $("word"), $("length"));
//        2> +I[a,b, a,b, 3]# a,b长度3
//        2> +I[a#b#c, a, 1] a#b#c 按井号依次拆分
//        2> +I[a#b#c, b, 1] ..
//        2> +I[a#b#c, c, 1] ..
//        2> +I[a#b, a, 1] a#b  按井号依次拆分
//        2> +I[a#b, b, 1]..
//        2> +I[c, c, 1] c

        Table result1 = orders
                .leftOuterJoinLateral(call(SplitFunction.class, $("e")).as("s",  "v"))
                .select($("a"), $("e"), $("s"),  $("v"));
        return result1;
//        1> +I[liji, a,b, a,b, 3]
//        1> +I[liji, a#b#c, a, 1]
//        1> +I[liji, a#b#c, b, 1]
//        1> +I[liji, a#b#c, c, 1]
//        1> +I[liji, a#b, a, 1]
//        1> +I[liji, a#b, b, 1]
//        1> +I[liji, c, c, 1]
//        1> +I[liji, c, c, 1]
    }



    //TODO
    private Table exFunction( Table orders) {
//        tEnv.registerFunction("myUdagg", new MyUdagg());
//        orders.groupBy("users")
//                .select(
//                        $("users"),
//                        call("myUdagg", $("points")).distinct().as("myDistinctResult")
//                );
        return null;
    }



    //TODO 变量太多
    private Table exFlatAggregate(Table orders) {
        //FlinkSQL 对表聚合函数支持并不是很好，这里使用TableAPI方式
        Table result1 = orders.groupBy($("a"))
                .flatAggregate(call(Top2.class, $("cnt")).as("value", "rank"))
                .select($("value"), $("rank"));
        Table result = orders
                .groupBy($("key"))
                .flatAggregate(call(Top2.class, $("a")).as("v", "rank"))
                .select($("key"), $("v"), $("rank"));
        return result1;
    }



    @FunctionHint(output = @DataTypeHint("ROW<word STRING, length INT>"))
    public static class SplitFunction extends TableFunction<Row> {

        public void eval(String str) {
            for (String s : str.split("#")) {
                // use collect(...) to emit a row
                collect(Row.of(s, s.length()));
            }
        }
    }
}
